package com.dc.dbus.transport.netty.client;

import com.dc.dbus.transport.commons.ClientSideReceiver;
import com.dc.dbus.transport.netty.Log;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
 
public class NettyClient {
	
	private final EventLoopGroup group;
	
	private final ClientChannelDisconnectHandler disconnectHandler;
	
	private final String nodeName;
	
    private final String host;
    private final int port;
    private final Channel channel;
    
    public NettyClient(EventLoopGroup group, String nodeName, String host, int port,
    		ClientSideReceiver clientSideReceiver,
    		ClientChannelDisconnectHandler disconnectHandler) throws InterruptedException {
    	this.group = group;
    	this.nodeName = nodeName;
        this.host = host;
        this.port = port;
        this.disconnectHandler = disconnectHandler;
        
        this.channel = start(clientSideReceiver);
    }
    
    public String getNodeName() {
		return nodeName;
	}

	public String getHost() {
		return host;
	}

	public int getPort() {
		return port;
	}

	protected Channel start(final ClientSideReceiver clientSideReceiver) throws InterruptedException {
    	
		Log.nettyLinkLogger.info("NettyClient.start: try link to:host:" + host + ":" + port + " ...");
		
        Bootstrap b = new Bootstrap();
        b.option(ChannelOption.TCP_NODELAY, true);
        b.group(group).channel(NioSocketChannel.class)
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                	
                    ChannelPipeline pipeline = ch.pipeline();
                    
                    pipeline.addLast("encode", new ObjectEncoder());
                    pipeline.addLast("decode", new ObjectDecoder(ClassResolvers.cacheDisabled(null)));
                    
                    pipeline.addLast(new NettyClientHandler(clientSideReceiver, disconnectHandler, NettyClient.this));
                    
                }
            });
        
        
        final ChannelFuture future = b.connect(host, port).sync();
 
        future.addListener(new ChannelFutureListener() {
 
            @Override
            public void operationComplete(ChannelFuture arg0) throws Exception {
                if (future.isSuccess()) {
                	Log.nettyLinkLogger.info("NettyClient.start: link to server:host:" + host + ":" + port + " SUCCESS");
                } else {
                	Log.nettyLinkLogger.error("NettyClient.start: link to server:host:" + host + ":" + port + " FAIL", future.cause());
                }
            }
        });

        return future.channel();
    }
 
    public Channel getChannel() {
        return channel;
    }
    
    public void close() {
    	channel.close();
    }
}